<?php

class Worker {
    // 监听的服务器 socket 资源
    protected $socket = NULL;

    // 连接建立时触发的回调函数
    public $onConnect = NULL;

    // 是否启用端口复用（SO_REUSEPORT）
    public $reusePort = 1;

    // 接收消息时触发的回调函数
    public $onMessage = NULL;

    // 子进程数量，默认为 10
    public $workerNum = 10;

    // 存放所有客户端连接的 socket 资源
    public $allSocket;

    // 服务器监听地址
    public $addr;

    // 子进程 PID 数组
    protected $worker_pid;

    // 主进程 PID
    protected $master_pid;

    /**
     * 构造函数，初始化服务器监听地址并记录主进程 PID
     */
    public function __construct($socket_address) {
        $this->addr = $socket_address;
        $this->master_pid = posix_getpid();
    }

    /**
     * 启动服务：创建子进程并开始监听
     */
    public function start() {
        $this->fork($this->workerNum); // 创建多个子进程
        $this->monitorWorkers(); // 监控子进程状态，处理信号
    }

    /**
     * 信号处理函数，用于接收 SIGUSR1 并执行 reload 操作
     */
    public function signalHandler($sigo) {
        switch ($sigo) {
            case SIGUSR1:
                $this->reload(); // 重启所有子进程
                echo "收到重启信号";
                break;
        }
    }

    /**
     * 创建指定数量的子进程
     *
     * @param int $worker_num 子进程数量
     */
    public function fork($worker_num) {
        for ($i = 1; $i < $worker_num; $i++) {
            $pid = pcntl_fork(); // 创建子进程

            if ($pid < 0) {
                exit('创建失败');
            } else if ($pid > 0) {
                // 父进程记录子进程 PID
                $this->worker_pid[] = $pid;
            } else {
                // 子进程负责接受客户端连接
                $this->accept();
                exit;
            }
        }
    }

    /**
     * 监控子进程，捕获信号并处理
     */
    public function monitorWorkers() {
        // 注册 SIGUSR1 信号处理函数
        pcntl_signal(SIGUSR1, array($this, 'signalHandler'), false);

        while (1) {
            \pcntl_signal_dispatch(); // 处理挂起的信号
            $status = 0;
            $pid = pcntl_wait($status, \WUNTRACED); // 等待子进程退出
            \pcntl_signal_dispatch(); // 再次处理可能的信号
        }
    }

    /**
     * 接受客户端连接并使用 Event 扩展处理事件循环
     */
    public function accept() {
        $opts = array(
            'socket' => array(
                'backlog' => 10240,
            ),
        );

        $context = stream_context_create($opts);
        stream_context_set_option($context, 'socket', 'so_reuseport', 1); // 端口复用
        stream_context_set_option($context, 'socket', 'so_reuseaddr', 1); // 地址复用

        // 创建 TCP 服务器 socket
        $this->socket = stream_socket_server($this->addr, $errno, $errstr, STREAM_SERVER_BIND | STREAM_SERVER_LISTEN, $context);
        stream_set_blocking($this->socket, false); // 设置为非阻塞模式

        $base = new \EventBase();
        $ss = $this->socket;

        // 监听服务器 socket 的可读事件（有新连接到达）
        $event2 = new \Event($base, $ss, \Event::PERSIST | \Event::READ | \Event::WRITE, function ($ss) {
            $clientSocket = stream_socket_accept($ss); // 接受新连接

            // 触发 onConnect 回调
            if (!empty($clientSocket) && is_callable($this->onConnect)) {
                call_user_func($this->onConnect, $clientSocket);
            }

            $base = new EventBase();

            // 监听客户端 socket 的可读/可写事件
            $event = new Event($base, $clientSocket, Event::PERSIST | Event::READ | Event::WRITE, function ($clientSocket) use (&$event) {
                $buffer = fread($clientSocket, 65535); // 读取数据

                if (empty($buffer)) {
                    if (!is_resource($clientSocket) || feof($clientSocket)) {
                        fclose($clientSocket); // 关闭连接
                    }
                }

                // 触发 onMessage 回调
                if (!empty($buffer) && is_callable($this->onMessage)) {
                    call_user_func($this->onMessage, $clientSocket, $buffer);
                }

                $event->del(); // 删除事件
            });

            $event->add(); // 添加事件到事件循环
            $base->loop(); // 启动事件循环
        });

        $event2->add(); // 添加事件到事件循环
        $base->loop(); // 启动事件循环
    }

    /**
     * 重启所有子进程
     */
    public function reload() {
        foreach ($this->worker_pid as $index => $pid) {
            posix_kill($pid, SIGKILL); // 终止旧子进程
            var_dump("杀掉的子进程", $pid);
            unset($this->worker_pid[$index]);
            $this->fork(1); // 重新启动一个子进程
        }
    }
}

function workerC10K(){
    // 创建 Worker 实例并绑定监听地址
    $worker = new Worker('tcp://0.0.0.0:9800');

    // 开启端口复用
    $worker->reusePort = true;

    // 客户端连接事件回调
    $worker->onConnect = function ($fd) {
        echo '连接事件触发',(int)$fd,PHP_EOL;
    };

    // 接收消息事件回调
    $worker->onMessage = function ($conn, $message) {
        echo "接收消息事件回调".PHP_EOL;
        $content = $message;
        $http_response = "HTTP/1.1 200 OK\r\n";
        $http_response .= "Content-Type: text/html;charset=UTF-8\r\n";
        $http_response .= "Connection: keep-alive\r\n";
        $http_response .= "Server: php socket server\r\n";
        $http_response .= "Content-length: " . strlen($content) . "\r\n\r\n";
        $http_response .= $content;

        fwrite($conn, $http_response); // 发送响应

        // 判断是否为空消息或连接已关闭
        if (empty($message)) {
            if (!is_resource($conn) || feof($message)) {
                fclose($conn); // 关闭连接
            }
        }
    };

    // 启动服务
    $worker->start();

}


function goC10K(){
    // Swoole 协程示例（简化版）
    Co\run(function () {
        $server = new Co\Socket(AF_INET, SOCK_STREAM, 0);
        $server->bind('0.0.0.0', 9800);
        $server->listen(65535);

        while (true) {
            //接受到新链接
            $conn = $server->accept();
            //创建协程任务把链接传递给协程任务异步处理不阻塞,继续走接受新连接
            go(function () use ($conn) {
                while (true) {
                    $data = $conn->recv();
                    if (!$data) break;
                    $conn->send("HTTP/1.1 200 OK\r\nContent-Length: " . strlen($data) . "\r\n\r\n" . $data);
                }
                $conn->close();
            });
        }
    });
}
goC10K();